package com.bosi.message.impl;

import com.alibaba.fastjson.JSON;
import com.bosi.message.Message;
import io.lettuce.core.RedisClient;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
import lombok.extern.slf4j.Slf4j;

/**
 * 基于redis的消息服务
 *
 * 将同一个游戏的消息广播到同一个通道
 */
@Slf4j
public class RedisMessageService extends SimpleMessageService {

    private final RedisClient client;
    private final String channel;
    private final StatefulRedisPubSubConnection<String, String> listenerConnection;
    private final RedisPubSubAsyncCommands<String, String> listenerCmd;
    private final StatefulRedisPubSubConnection<String, String> publishConnection;
    private final RedisPubSubAsyncCommands<String, String> publishCmd;

    public RedisMessageService(RedisClient client, String channel) {
        this.client = client;
        this.channel = channel;
        this.listenerConnection = client.connectPubSub();
        this.publishConnection = client.connectPubSub();
        this.publishCmd = publishConnection.async();
        this.listenerCmd = listenerConnection.async();

        //添加监听通道
        listenerCmd.subscribe(this.channel);
        //添加消息监听器
        listenerConnection.addListener(new RedisPubSubAdapter<String,String>(){
            @Override
            public void message(String channel, String text) {
                Message message = JSON.parseObject(text, Message.class);
                RedisMessageService.super.publish(message);
            }
        });
    }

    @Override
    public void publish(Message message) {
        publishCmd.publish(this.channel,JSON.toJSONString(message));
    }
}
